package com.yuan.key_code;

import java.io.IOException;
import java.io.PrintStream;
import java.net.URI;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
 
/**
 * 过滤掉文件名满足特定条件的文件 
 */
class MyPathFilter implements PathFilter {
     String reg = null; 
     MyPathFilter(String reg) {
          this.reg = reg;
     }
     public boolean accept(Path path) {
        if (!(path.toString().matches(reg)))
            return true;
        return false;
    }
}
/***
 * 利用FSDataOutputStream和FSDataInputStream合并HDFS中的文件
 */
public class MergeFile {
    Path inputPath = null; //待合并的文件所在的目录的路径
    Path outputPath = null; //输出文件的路径
    public MergeFile(String input, String output) {
        this.inputPath = new Path(input);
        this.outputPath = new Path(output);
    }
    public void doMerge() throws IOException {
        Configuration conf = new Configuration();

        //设置HDFS的默认文件系统URI，固定写法
        conf.set("fs.defaultFS","hdfs://localhost:9000");
        conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");

        //!获取输入路径和输出路径的文件系统对象
        FileSystem fsSource = FileSystem.get(URI.create(inputPath.toString()), conf);
        FileSystem fsDst = FileSystem.get(URI.create(outputPath.toString()), conf);

        //下面过滤掉输入目录中后缀为.abc的文件数组
        FileStatus[] sourceStatus = fsSource.listStatus(inputPath,
                new MyPathFilter(".*\\.abc")); //过滤规则：排除后缀为.abc的文件

        //创建输出文件的写入流
        FSDataOutputStream fsdos = fsDst.create(outputPath);

        //创建打印流，将内容同时输出到控制台
        PrintStream ps = new PrintStream(System.out);

        //下面分别读取过滤之后的每个文件的内容，并输出到同一个文件中
        for (FileStatus sta : sourceStatus) {
            //下面打印后缀不为.abc的文件的路径、文件大小
            System.out.print("路径：" + sta.getPath() + "    文件大小：" + sta.getLen()
                    + "   权限：" + sta.getPermission() + "   内容：");
            
            //下面打开当前文件的输入流
            FSDataInputStream fsdis = fsSource.open(sta.getPath());
            byte[] data = new byte[1024];//缓冲区，每次读取1024字节
            int read = -1;
 
            //循环读取文件内容
            while ((read = fsdis.read(data)) > 0) {
                ps.write(data, 0, read);//将内容写入控制台
                fsdos.write(data, 0, read);//!将内容写入输出文件
            }
            fsdis.close();//关闭当前文件的输入流          
        }
        ps.close();//关闭打印流
        fsdos.close();//关闭输出文件的写入流
    }
    public static void main(String[] args) throws IOException {
        //! hdfs://localhost:9000指的是ubuntu运行的伪分布HDFS文件系统而非本地文件系统
        MergeFile merge = new MergeFile(
                "hdfs://localhost:9000/user/hadoop/",
                "hdfs://localhost:9000/user/hadoop/merge.txt");
        merge.doMerge();
    }
}
